package com.navinfo.opentsp.platform.computing.serializer;

import com.google.common.base.Charsets;
import com.navinfo.opentsp.platform.computing.util.DateUtils;
import com.navinfo.opentsp.platform.location.protocol.gbv6.hangzhou.LCHangzhouDischarge;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.serialization.EventSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.OutputStream;

/**
* hangzhouDischarge信息
* HDFS_Sink的hangzhouDischarge序列化器，拼接成字段间通过\001分割的字符串
*/
public class HDFSHangzhouDischargeSerializer implements EventSerializer {
    private static final Logger logger = LoggerFactory.getLogger(HDFSHangzhouDischargeSerializer.class);

    private final String APPEND_NEWLINE = "appendNewline";
    private final boolean appendNewline;
    private final String splitStr = "\001";
    private final OutputStream out;

    private HDFSHangzhouDischargeSerializer(OutputStream out, Context ctx) {
        this.appendNewline = ctx.getBoolean(APPEND_NEWLINE, true);
        this.out = out;
    }

    @Override
    public boolean supportsReopen() {
        return false;
    }
    @Override
    public void afterCreate() {}
    @Override
    public void afterReopen() {}
    @Override
    public void beforeClose() {}
    @Override
    public void flush() throws IOException {
        this.out.flush();
    }
    @Override
    public void write(Event event) throws IOException {
        try {
            //拼写落盘到HDFS的字符串。字段间分隔符是\001，null通过\N表示
            LCHangzhouDischarge.HangzhouDischarge hangzhouDischarge = LCHangzhouDischarge.HangzhouDischarge.parseFrom(event.getBody());
            StringBuilder hangzhouDischargeStr = new StringBuilder(5000);
                    hangzhouDischargeStr.append(hangzhouDischarge.getTime()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getLongitude()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getLatitude()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getTemperature()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getAtmosphericPressure()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getSpeed()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getMile()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getAccelerator()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getHandBrake()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getFootBrake()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getAcc()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getMil()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getFuelLevel()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getIntakeManifoldPressure()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getIntakeManifoldTemp()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getOilPressuer()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getVoltage()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getTotalFuelConsumption()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getUreaVoltage()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getFuelTemp()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getCoolantTemp()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getIntakeFlow()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getIntakeTemp()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getRealTimeOilConsumption()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getEngineTorMode()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getActualEnginePercentTor()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getFrictionEnginePercentTor()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getIdlingTor()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getActualInnerPercentTor()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getReferenceTor()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getRotation()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getRealtimeFuelInjection()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getEngineStartMode()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getCumulativeRunningTime()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getCumulativeRotation()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getActiveLimitType()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getFuelQuality()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getUreaLevel()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getUreaTemp()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getActualBlueSpout()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getIntakeOxynitrideValue()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getAftertreatOxynitrideContr()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getTotalUreaConsumption()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getScrSubsystemStatus()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getUreaPumpDutyfactor()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getUreaSprayerDutyfactor()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getUreaLevelSignalCheck()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getAftertreatUpstreamExhaustTemp()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getCatalystDownStreamTemp()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getUpstreamExhausTempSignalCheck()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getAftertreatDewpointSts()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getTailPipeOxynitrideDewPointSts()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getNoxSensorSignalCheck()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getIntakeOxygenContent()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getExhaustOxygenContent()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getPumpRotation()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getPumpStatus()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getIntakePressure()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getIntakeTempPoc()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getExhaustPressure()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getExhaustTemp()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getCarrierPressureDifferential()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getWasteGasQuantityFlow()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getAftertreatRealtimeInjection()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getTailpipeRealtimeOxynitrideContr()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getPemsNoxExhaust()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getEcuTamperDetection()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getTerminalId()).append(splitStr);
                    hangzhouDischargeStr.append(hangzhouDischarge.getSerialNumber()).append(splitStr);

            //添加字段只能在分割线之上添加,注意时间分区这块，终端传的时候是秒还是毫秒
            //------------------------------------------分割线----------------------------------------
            hangzhouDischargeStr.append(DateUtils.format(hangzhouDischarge.getTime(), "yyyyMMdd"));

            //将数据拼接后写入hdfs
            this.out.write(hangzhouDischargeStr.toString().getBytes(Charsets.UTF_8));
            if (this.appendNewline) {
                this.out.write(10);
            }
        }catch (Exception e){
            logger.error("执行HDFS的HDFSHangzhouDischargeSerializer拦截器失败：" + e.getMessage(), e);
        }
    }

    public static class Builder implements EventSerializer.Builder {
        public Builder() { }
        @Override
        public EventSerializer build(Context context, OutputStream out) {
            return  new HDFSHangzhouDischargeSerializer(out, context);
        }
    }
}
